package com.gitee.sparta.core.cluster.nio;

import cn.hutool.core.collection.CollUtil;
import com.gitee.sparta.core.bean.cluster.ClusterProtocol;
import com.gitee.sparta.core.cluster.AbstractClusterFollower;
import com.gitee.sparta.core.cluster.AbstractClusterLeader;
import com.gitee.sparta.core.cluster.ClusterCache;
import com.gitee.sparta.core.cluster.ClusterCommonCode;
import com.gitee.sparta.core.cluster.balance.LoadBalance;
import com.gitee.sparta.core.cluster.strategy.StrategyFactory;
import com.gitee.sparta.core.property.SpartaCoreProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.context.ApplicationEventPublisher;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

/**
 * follower nio
 *
 * @author zyred
 * @since v 1.0.0
 * @deprecated 弃用 nio 作为通讯组件
 */
@Slf4j
@Deprecated
public class NioClusterFollower extends AbstractClusterFollower {

    private Selector selector;
    private SocketChannel socketChannel;
    private final ClusterCommonCode ccc;
    private final AbstractClusterLeader leader;

    /**
     * follower 节点启动器
     *  @param eventPublisher        事件发布器
     * @param client                redisson client
     * @param properties            配置文件
     * @param loadBalance           负载均衡
     * @param leader                leader
     */
    public NioClusterFollower(ApplicationEventPublisher eventPublisher,
                              RedissonClient client,
                              SpartaCoreProperties properties,
                              LoadBalance loadBalance,
                              AbstractClusterLeader leader,
                              ClusterCommonCode ccc) {
        super(eventPublisher, client, properties, loadBalance);
        this.ccc = ccc;
        this.leader = leader;
    }

    @Override
    public void buildFollower(String host, int port) throws IOException {
        selector = Selector.open();
        socketChannel = SocketChannel.open(new InetSocketAddress(host, port));
        socketChannel.configureBlocking(false);
        socketChannel.register(selector, SelectionKey.OP_READ);
        this.start();

        this.ccc.firstConnectionLeader(ClusterCache.getServerId());
        log.info("[Sparta] Nio connect leader success, host : {}, port : {}", host, port);
    }

    @SuppressWarnings("all")
    public void start() {
        new Thread(() -> {
            while (!Thread.interrupted()) {
                try {
                    selector.select();
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> it = selectionKeys.iterator();
                    while (it.hasNext()) {
                        SelectionKey key = it.next();
                        it.remove();
                        if (key.isReadable()) {
                            SocketChannel self = (SocketChannel) key.channel();
                            this.handlerReads(self);
                            self.register(selector, SelectionKey.OP_READ);
                        }
                    }
                } catch (IOException ex) {
                    // 断开 leader 的连接
                    log.error("[Sparta] Leader disconnection, exception : {}", ex.getMessage());
                    this.ccc.promotionLeader(leader, this);
                }
            }

            log.info("[Sparta] Follower destroy ....");
            try {
                socketChannel.close();
            } catch (IOException ex) {
                log.error("[Sparta] Follower close error, exception : {}", ex.getMessage(), ex);
            }
        }, "nio-follower").start();
    }


    /**
     * 处理 leader 发送过来的数据
     * <p>
     *
     * @param self 通道
     * @throws IOException IOException
     */
    private void handlerReads(SocketChannel self) throws IOException {
        List<ClusterProtocol> protocolList = HandlerReadWriteHelper.readProtocol(self);
        if (CollUtil.isEmpty(protocolList)) {
            return;
        }

        for (ClusterProtocol protocol : protocolList) {
            StrategyFactory.getStrategy(protocol.getCmd())
                    .handle(null, self, protocol);
        }
    }

    @Override
    public void closeForOldLeader() throws IOException {
        this.socketChannel.close();
        log.info("[Sparta] Close old leader nio connection.");
    }

    @Override
    protected void doWriteLeader(ClusterProtocol protocol) throws Exception {
        HandlerReadWriteHelper.writeProtocol(protocol, this.socketChannel);
    }
}
